package com.relytosoft.mizar.analyze.face.services.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.relytosoft.dubhe.common.kafka.producer.KafkaOpMsg;
import com.relytosoft.dubhe.common.redis.utils.RedisUtil;
import com.relytosoft.mizar.analyze.base.constant.AyCacheConstants;
import com.relytosoft.mizar.analyze.base.constant.MediaMqConstants;
import com.relytosoft.mizar.analyze.face.api.entity.AyFaceDisposition;
import com.relytosoft.mizar.analyze.face.services.service.AyFaceDispositionService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Map;

;

/**
 * @author lj
 * @date 2022/5/19 11:52
 */
@Component
public class FaceDispositionConsumer {
    protected Logger log = LoggerFactory.getLogger(this.getClass());

    @Resource
    private AyFaceDispositionService ayFaceDispositionService;

    @KafkaListener(id = "consumer-mizar-face-disposition-sync", topics = MediaMqConstants.FACE_ANALYZE_DISPOSITION_ID_TOPIC,
            containerFactory = "singleContainerFactory")
    public void processSingleMessage(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        try {
            KafkaOpMsg kafkaOpMsg = JSON.parseObject(record.value().toString(), KafkaOpMsg.class);
            log.info("线程:{} 消费->{}", Thread.currentThread(), kafkaOpMsg.getOpType());
            String data = kafkaOpMsg.getData();
            Map<String, Object> map = JSONObject.parseObject(data, new TypeReference<Map<String, Object>>() {
            });
            //自己系统布控任务ID
            Integer id = (Integer) map.get("id");
            //三方布控任务ID
            String dispositionId = (String) map.get("dispositionId");
            //根据任务ID查询
            AyFaceDisposition ayFaceDisposition = ayFaceDispositionService.getById(id);
            ayFaceDisposition.setDispositionId(dispositionId);
            //当获取到返回的布控ID说明任务已经开始
            ayFaceDisposition.setStatus(1);
            ayFaceDisposition.setDefenseStatus(1);
            ayFaceDisposition.setBeginTime(LocalDateTime.now());
            //修改状态并存入缓存和redis
            boolean isConsume = ayFaceDispositionService.updateById(ayFaceDisposition);
            RedisUtil.set(AyCacheConstants.CONTROL_FACE_CONDITION + dispositionId, JSON.toJSONString(ayFaceDisposition));
            if (isConsume) {
                ack.acknowledge();
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error(e.getMessage());
        }
    }
}
